package com.chukun.flink.stream.window.process.function;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @author chukun
 * @version 1.0.0
 * @description 自定义process函数处理
 * @createTime 2022年05月23日 00:38:00
 */
public class ExportProcessWindow extends ProcessWindowFunction<Tuple3<String, Integer, String>, Tuple2<Long, Tuple3<String, Integer, String>>, String, TimeWindow> {
    @Override
    public void process(String key, 
                        ProcessWindowFunction<Tuple3<String, Integer, String>, Tuple2<Long, Tuple3<String, Integer, String>>, String, TimeWindow>.Context context, 
                        Iterable<Tuple3<String, Integer, String>> minReadings, 
                        Collector<Tuple2<Long, Tuple3<String, Integer, String>>> collector) throws Exception {
        // 窗口函数的迭代器中只有一个值处理
        Iterator<Tuple3<String, Integer, String>> it = minReadings.iterator();
        if (it.hasNext()) {
            // 获取ReduceFunction窗口函数输出的最终结果
            Tuple3<String, Integer, String> min = it.next();
            long start = context.window().getStart();
            collector.collect(Tuple2.of(start, min));
        }
    }
}
